package com.pig4cloud.pigx.contract.websocket.subscribe;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.pig4cloud.pigx.common.websocket.holder.SessionKeyGenerator;
import com.pig4cloud.pigx.contract.config.contract.ContractConfig;
import com.pig4cloud.pigx.contract.dto.AppContractCurrentOrderConditionDTO;
import com.pig4cloud.pigx.contract.dto.AppContractCurrentPositionConditionDTO;
import com.pig4cloud.pigx.contract.enums.ContractOrderEnum;
import com.pig4cloud.pigx.contract.enums.ContractPositionEnum;
import com.pig4cloud.pigx.contract.mapper.ContractOrderMapper;
import com.pig4cloud.pigx.contract.mapper.ContractPositionMapper;
import com.pig4cloud.pigx.contract.mongo.entity.ContractSymbolMongo;
import com.pig4cloud.pigx.contract.utils.ContractUtils;
import com.pig4cloud.pigx.contract.vo.AbstractTopicVO;
import com.pig4cloud.pigx.contract.vo.AppContractCurrentOrderVO;
import com.pig4cloud.pigx.contract.vo.AppContractCurrentPositionVO;
import com.pig4cloud.pigx.contract.vo.AppContractCurrentStopOrderVO;
import com.pig4cloud.pigx.contract.vo.Contract24hTradeVo;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 */
@Slf4j
@Component
public class MergeContract extends DataListAbstractTopic {

    private final SessionKeyGenerator sessionKeyGenerator;
    private final ContractOrderMapper contractOrderMapper;
    private final ContractUtils utils;
    private final ContractPositionMapper contractPositionMapper;
    private final MongoTemplate mongo;
    private final ContractConfig contractConfig;
    private final RedisTemplate redis;
    @Autowired
    private ThreadPoolTaskExecutor websocketThreadPool;

    protected String TOPICNAME = "mergeContract";

    public MergeContract(SessionKeyGenerator sessionKeyGenerator, ContractOrderMapper contractOrderMapper, ContractUtils utils, ContractPositionMapper contractPositionMapper, MongoTemplate mongo, ContractConfig contractConfig, RedisTemplate redis) {
        this.sessionKeyGenerator = sessionKeyGenerator;
        this.contractOrderMapper = contractOrderMapper;
        this.utils = utils;
        this.contractPositionMapper = contractPositionMapper;
        this.mongo = mongo;
        this.contractConfig = contractConfig;
        this.redis = redis;
        init();
    }

    private void init() {
        wsPush();
    }

    private void wsPush() {
        // 获取线程处理数量
//		Integer handleSize = contractConfig.getWebsocket().getThreadHandleSizeMergeContract();
        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        service.scheduleAtFixedRate(() -> {
            try {
				/*// 收集未运行的会话，并修改运行状态为运行中
				List<Map.Entry<WebSocketSession, AbstractTopicVO>> entryList = subList.entrySet().stream()
						.filter(entry -> startRun(entry.getValue()))
						.collect(Collectors.toList());

				for (int i = 0; i < entryList.size(); ) {
					int startIndex = i;
					i += handleSize;
					// 下标越界
					if (i >= entryList.size()) {
						i = entryList.size();
					}

					// 截取
					List<Map.Entry<WebSocketSession, AbstractTopicVO>> partList = entryList.subList(startIndex, i);
					// 发送消息
					sendMessage(partList);
				}*/

                for (Map.Entry<WebSocketSession, AbstractTopicVO> entry : subList.entrySet()) {
                    if (!entry.getKey().isOpen()) {
                        subList.remove(entry.getKey());
                        log.info("ws 移除已关闭的session");
                        continue;
                    }
					/*Object userId = sessionKeyGenerator.sessionKey(entry.getKey());
					if ("9000".equals(userId.toString())) {
						continue;
					}*/

                    this.sendMessage(entry);
                }

            } catch (Throwable e) {
                e.printStackTrace();
                log.info("push merge contract err:" + e.getMessage());
            }
        }, 0, 2000, TimeUnit.MILLISECONDS);
    }

    public void sendMessage(Map.Entry<WebSocketSession, AbstractTopicVO> entry) {
        if (!entry.getKey().isOpen()) {
            subList.remove(entry.getKey());
            log.info("ws 移除已关闭的session");
            return;
        }
        final WebSocketSession session = entry.getKey();
        AbstractTopicVO param = entry.getValue();

        if (startRun(param)) {
            // 开启线程，发送数据
            websocketThreadPool.execute(new mergeContractTask(entry));
        }

		/*try {
			if (!entry.getKey().isOpen()) {
				subList.remove(entry.getKey());
				log.info("ws 移除已关闭的session");
				return;
			}
			final WebSocketSession session = entry.getKey();

			Object userId = sessionKeyGenerator.sessionKey(session);
			AbstractTopicVO param = entry.getValue();
			List<String> dataList = param.getDataList();
			// 当前委托
			List<AppContractCurrentOrderVO> order = getOrder(userId, dataList);
			// 止盈止损
			List<AppContractCurrentStopOrderVO> stopOrder = getStopOrder(userId, dataList);
			// 当前持仓sys_dict
			List<AppContractCurrentPositionVO> position = getPosition(userId, dataList);
			// 资金费率 + 涨跌幅
			Contract24hTradeVo riseAndFall = getRiseAndFall(dataList);
			// 四合一
			JSONObject object = new JSONObject() {{
				put("order", order);
				put("stopOrder", stopOrder);
				put("position", position);
				put("detailed", riseAndFall);
			}};

			session.sendMessage(getTextMessage(dataList, object));

		} catch (Throwable e) {
			log.info("push merge contract err:" + e.getMessage());
		} finally {
			// 清除symbol缓存
			ContractUtils.removeSymbolThreadCache();
		}*/
    }

    /**
     * 推送用户合约数据线程
     *
     * @since 2022/9/27 20:23
     */
    @Data
    @AllArgsConstructor
    private class mergeContractTask implements Runnable {

        private final Map.Entry<WebSocketSession, AbstractTopicVO> entry;

        /**
         * 推送用户合约数据
         *
         * @since 2022/9/27 20:23
         */
        @Override
        public void run() {
            final WebSocketSession session = entry.getKey();

            Object userId = sessionKeyGenerator.sessionKey(session);
            AbstractTopicVO param = entry.getValue();

            try {
                List<String> dataList = param.getDataList();
                // 当前委托
                List<AppContractCurrentOrderVO> order = getOrder(userId, dataList);
                // 止盈止损
                List<AppContractCurrentStopOrderVO> stopOrder = getStopOrder(userId, dataList);
                // 当前持仓sys_dict
                List<AppContractCurrentPositionVO> position = getPosition(userId, dataList);
                // 资金费率 + 涨跌幅
                Contract24hTradeVo riseAndFall = getRiseAndFall(dataList);
                // 四合一
                JSONObject object = new JSONObject() {{
                    put("order", order);
                    put("stopOrder", stopOrder);
                    put("position", position);
                    put("detailed", riseAndFall);
                }};

                session.sendMessage(getTextMessage(dataList, object));

            } catch (Throwable e) {
                e.printStackTrace();
                log.info("push merge contract err:" + e.getMessage());
            } finally {
                // 清除symbol缓存
                ContractUtils.removeSymbolThreadCache();
                endRun(param);
            }
        }
    }

    private Contract24hTradeVo getRiseAndFall(List<String> dataList) {
        String symbolId = dataList.get(1);
        Query query = new Query();
        query.addCriteria(Criteria.where("_id").is(Long.valueOf(symbolId)));
        ContractSymbolMongo one = mongo.findOne(query, ContractSymbolMongo.class);
        Contract24hTradeVo vo = new Contract24hTradeVo();
        BeanUtils.copyProperties(one, vo);
        vo.setRiseAndFallRange(vo.getRiseAndFallRange().setScale(2, RoundingMode.HALF_UP));
        vo.setFundingRate(vo.getFundingRate().multiply(BigDecimal.valueOf(100)).setScale(4, RoundingMode.HALF_UP));
        utils.setScaleBySymbol(vo, one.getName());
        return vo;
    }

    public List<AppContractCurrentStopOrderVO> getStopOrder(Object userId, List<String> dataList) {
        return new ArrayList<>(0);
    }

    public List<AppContractCurrentOrderVO> getOrder(Object userId, List<String> dataList) throws JsonProcessingException {
        AppContractCurrentOrderConditionDTO condition = new AppContractCurrentOrderConditionDTO();
        condition.setUserId(Long.valueOf(userId.toString()));
        // 币对id
        if (dataList.size() > 0) {
            String symbolId = dataList.get(1);
            String isAll = dataList.get(2);
            if (Strings.isNotEmpty(symbolId) && Strings.isNotEmpty(isAll)) {
                condition.setSymbolId(Long.valueOf(symbolId));
            }
        }

        // 如果不是只查询总数
        List<AppContractCurrentOrderVO> voList = contractOrderMapper.appListCurrentOrder(condition);
        // 遍历
        voList.forEach(vo -> {
            vo.setCurrencyNum(vo.getNum().multiply(vo.getPrice()));
            if (vo.getDealNum() != null && vo.getDealAvg() != null) {
                vo.setCurrencyDealNum(vo.getDealNum().multiply(vo.getDealAvg()));
            }
            // 翻译字典项
            vo.setType(ContractOrderEnum.Type.findName(vo.getType()));
            vo.setPositionMode(ContractPositionEnum.Mode.findName(vo.getPositionMode()));
            vo.setSide(ContractOrderEnum.Side.findName(vo.getSide()));
            vo.setOperationType(ContractOrderEnum.OperationType.findName(vo.getOperationType()));
            vo.setStatus(ContractOrderEnum.Status.findName(vo.getStatus()));
            // 张数
            ContractSymbolMongo one = mongo.findOne(Query.query(Criteria.where("name").is(vo.getSymbolName())), ContractSymbolMongo.class);
            if (one != null) {
                vo.setSheet(one.getSheet());
            }
            // 设置精度
            utils.setScaleBySymbol(vo, vo.getSymbolName());
        });

        return voList;
    }

    public List<AppContractCurrentPositionVO> getPosition(Object userId, List<String> dataList) {
        AppContractCurrentPositionConditionDTO condition = new AppContractCurrentPositionConditionDTO();
        condition.setUserId(Long.valueOf(userId.toString()));
        // 币对id
        if (dataList.size() > 0) {
            String symbolId = dataList.get(1);
            String isAll = dataList.get(2);
            if (Strings.isNotEmpty(symbolId) && Strings.isNotEmpty(isAll)) {
                condition.setSymbolId(Long.valueOf(symbolId));
            }
        }

        // 查询数据
        List<AppContractCurrentPositionVO> voList = contractPositionMapper.appListCurrentPosition(condition);
        // 遍历
        voList.forEach(vo -> {
            // 计算未实现盈亏、收益率，设置标记价
            utils.computeFloatProfit(vo);
            // 计算页面保证金和保证金率
            utils.computeMarginRate(vo);

            // 计算可用数量（USDT）
            vo.setAvailableMainCurNum(vo.getAvailableNum().multiply(vo.getOpenAvgPrice()));
            vo.setUNowNum(vo.getNowNum().multiply(vo.getMarkPrice()));
            vo.setUAvailableNum(vo.getAvailableNum().multiply(vo.getMarkPrice()));
            vo.setUClosingNum(vo.getClosingNum().multiply(vo.getMarkPrice()));
            // 张数
            ContractSymbolMongo one = mongo.findOne(Query.query(Criteria.where("name").is(vo.getSymbolName())), ContractSymbolMongo.class);
            if (one != null) {
                vo.setSheet(one.getSheet());
            }

            // 设置精度
            utils.setScaleBySymbol(vo, vo.getSymbolName());

            // 未实现盈亏先默认2位精度
            vo.setFloatProfit(vo.getFloatProfit().setScale(2, RoundingMode.HALF_UP));

            // 翻译字典项
            vo.setMode(ContractPositionEnum.Mode.findName(vo.getMode()));
            vo.setSide(ContractOrderEnum.Side.findName(vo.getSide()));
            vo.setStatus(ContractPositionEnum.Status.findName(vo.getStatus()));
        });

        return voList;
    }
}
